//
// Created by edward on 23-5-17.
//

#include "terminal.h"
#include "utils.h"
#include <pb_encode.h>
#include <pb_decode.h>
#include "message.pb.h"
#include <signal.h>

#define MAX_EVENT_NUMBER 1024

int terminalNum;
int id, controller_id;
struct Socket sockets[MAX_TERMINAL_NUM];
int active_connect[MAX_TERMINAL_NUM];
int active_connect_len;

int pipefd[2];  //管道
int epfd;
struct epoll_event events[MAX_EVENT_NUMBER];

typedef void(*Handler)(void *);
struct AlarmEvent {
    Handler handler;
    void *p;
};
#define ALARM_EVENTS_NUM 20
struct AlarmEvent alarmEvents[ALARM_EVENTS_NUM], tmpAlarmEvents[ALARM_EVENTS_NUM];
int alarmEventNum = 0, tmpAlarmEventNum;
#define ALARM_TIME 1

int confirm_cnt = 0;
int ready_cnt = 0;

char buf[BUFSIZ];

void encode_message(int receiverId, int cmdId, int cmdData) {
    size_t message_length;
    bool status;

    Message message = Message_init_zero;
    /* Create a stream that will write to our buffer. */
    pb_ostream_t stream = pb_ostream_from_buffer(buf, BUFSIZ);

    /* Fill in the message */
    message.sender_id = id;
    message.receiver_id = receiverId;
    message.command_id = cmdId;
    message.command_data = cmdData;

    status = pb_encode(&stream, Message_fields, &message);
    message_length = stream.bytes_written;

    if (!status) {
        log_fatal("encoding failed: %s", PB_GET_ERROR(&stream));
        exit(1);
    }

    log_debug("[%d] send message: {%d -> %d : %d - %d}", id, id, receiverId, cmdId, cmdData);
    Write(sockets[receiverId].sockfd, buf, message_length);
}

void send_ready(void *) {
    if (sockets[controller_id].sockfd == 0) {
        //说明与初始控制终端的连接尚未成功
        log_trace("[%d] wait control terminal to send ready message", id);
        alarmEvents[alarmEventNum++].handler = send_ready;
    } else {
        encode_message(controller_id, 2, 0);
    }
}

void recv_message(int sockfd, int senderId, int cmdId, int cmdData) {
    switch (cmdId) {
        case 0:
        {
            //收到消息的ACK
            switch (cmdData) {
                case 1:
                {
                    //确认编号的ACK
                    if (++confirm_cnt == active_connect_len) {
                        //如果所有主动连接的对端都发送了ACK 1，那么说明连接建立已经成功，通知初始控制终端
                        if (controller_id != id) {
                            send_ready(NULL);
                            if (alarmEventNum > 0) {
                                //定时
                                alarm(ALARM_TIME);
                            }
                        }
                        else {
                            log_info("[%d] terminal %d has been ready", id, id);
                            if (++ready_cnt == terminalNum) {
                                //所有终端已经就绪
                                log_info("[%d] all terminal has been ready", id);
                            } else {
                                log_debug("[%d]  %d terminal has not been ready", id, terminalNum - ready_cnt);
                            }
                        }
                    }
                }
                    break;
            }
        }
            break;
        case 1:
        {
            //确认编号，根据senderId调整编号
            for (int i = 0; i < terminalNum; ++i) {
                if (sockets[i].sockfd == sockfd) {
                    if (senderId != i) {
                        //交换到正确的编号
                        swap(&sockets[i], &sockets[senderId]);
                    }
                    break;
                }
            }
            //发送ACK
            encode_message(senderId, 0, 1);
        }
            break;
        case 2:
        {
            //收到向控制终端发送的消息
            switch (cmdData) {
                case 0:
                {
                    //应该只有控制终端可以收到准备就绪消息
                    assert(id == controller_id);
                    //终端已经就绪
                    log_info("[%d] terminal %d has been ready", id, senderId);
                    if (++ready_cnt == terminalNum) {
                        //所有终端已经就绪
                        log_info("[%d] all terminal has been ready", id);
                    } else {
                        log_debug("[%d]  %d terminal has not been ready", id, terminalNum - ready_cnt);
                    }
                }
                    break;
            }
        }
            break;
    }
}

void decode_message(int sockfd, int bufSize) {
    /* 创建一个输入流 */
    pb_istream_t stream = pb_istream_from_buffer(buf, bufSize);
    bool status;
    /* 在循环中解码消息，直到没有更多的消息 */
    while (stream.bytes_left > 0) {
        Message message = Message_init_zero;
        status = pb_decode(&stream, Message_fields, &message);
        if (!status) {
            log_fatal("decoding failed: %s", PB_GET_ERROR(&stream));
        }
        int senderId = message.sender_id;
        int receiverId = message.receiver_id;
        int cmdId = message.command_id;
        int cmdData = message.command_data;
        log_debug("[%d] recv message: {%d -> %d : %d - %d}", id, senderId, receiverId, cmdId, cmdData);
        assert(receiverId == id);   //接收者id应该是当前终端id，否则应该是发错了

        recv_message(sockfd, senderId, cmdId, cmdData);
    }
}


int set_nonblocking(int sockfd) {
    int old_option = fcntl(sockfd, F_GETFL);
    int new_option = old_option | O_NONBLOCK;
    fcntl(sockfd, F_SETFL, new_option);
    return old_option;
}

void set_opt(int sockfd, int opt) {
    fcntl(sockfd, F_SETFL, opt);
}

void epoll_add_fd2read(int sockfd, bool is_ET) {
    struct epoll_event event;
    event.data.fd = sockfd;
    event.events = EPOLLIN;
    if (is_ET) {
        event.events |= EPOLLET;
        set_nonblocking(sockfd);
    }
    Epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &event);
}

void epoll_add_fd2write(int sockfd, bool is_ET) {
    struct epoll_event event;
    event.data.fd = sockfd;
    event.events = EPOLLOUT;
    if (is_ET) {
        event.events |= EPOLLET;
        set_nonblocking(sockfd);
    }
    Epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &event);
}

void epoll_del_fd2read(int  sockfd) {
    struct epoll_event event;
    event.data.fd = sockfd;
    event.events = EPOLLIN;
    Epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, &event);
}

void epoll_del_fd2write(int  sockfd) {
    struct epoll_event event;
    event.data.fd = sockfd;
    event.events = EPOLLOUT;
    Epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, &event);
}

void epoll_exit() {
    for (int i = 0; i < terminalNum; ++i) {
        if (sockets[i].sockfd != 0) {
            Close(sockets[i].sockfd);
        }
    }
    Close(pipefd[0]);
    Close(pipefd[1]);
    log_info("[%d] exit", id);
    exit(0);
}


/*
 * epoll_wait返回处理信号
 */
void handle_signal(int fd) {
    int ret = Read(fd, buf, BUFSIZ - 1);
    for (int i = 0; i < ret; ++i) {
        switch (buf[i]) {
            case SIGHUP:
                log_trace("[%d] signal caught SIGHUP", id);
                break;
            case SIGCHLD:
                log_trace("[%d] signal caught SIGCHLD", id);
                break;
            case SIGTERM:
                log_trace("[%d] signal caught SIGTERM", id);
                break;
            case SIGINT:
                log_trace("[%d] signal caught SIGINT", id);
                epoll_exit();
                break;
            case SIGQUIT:
                log_trace("[%d] signal caught SIGQUIT", id);
                break;
            case SIGALRM:
                log_trace("[%d] signal caught SIGALRM", id);

                // 先进行拷贝，允许在处理定时事件的时候继续其他定时事件，此次添加的定时事件会在下次定时结束运行
                tmpAlarmEventNum = alarmEventNum;
                for (int i = 0; i < tmpAlarmEventNum; ++i) {
                    tmpAlarmEvents[i] = alarmEvents[i];
                }
                alarmEventNum = 0;
                for (int i = 0; i < tmpAlarmEventNum; ++i) {
                    tmpAlarmEvents[i].handler(tmpAlarmEvents[i].p);
                }

                // 如果处理完了所有定时事件，且没有增加新的定时事件，关闭定时，否则重新设置定时
                if (alarmEventNum > 0) {
                    alarm(ALARM_TIME);
                } else {
//                    alarm(0);
                }
                break;
        }
    }
}

/*
 * epoll_wait返回处理连接事件
 */
void accept_client(int lfd) {
    //客户端连接
    int clientIdx = 0;
    for (int i = 0; i < terminalNum; ++i) {
        if (sockets[i].sockfd == 0) {
            clientIdx = i;
            break;
        }
    }
    struct Socket* sckt = &sockets[clientIdx];
    easy_accept(lfd, sckt);
    log_trace("[%d] terminal {%s : %d} in %d connected", id, sckt->ip, sckt->port, clientIdx);
    epoll_add_fd2read(sckt->sockfd, true);

}

/*
 * epoll_wait返回处理可读事件
 */
void read_buf(int fd) {
    //客户端发送数据
    // 以et模式读取数据
    while (true) {
        int ret = Read(fd, buf, BUFSIZ - 1);
        if (ret > 0) {
            buf[ret] = '\0';
//            log.print("[client " + clientSocket.ip + ":" + to_string(clientSocket.port) + "]\n{" + buf +"}");
//            printf("terminal %d [%s : %d] sent: %s\n", clientSocket->idx, clientSocket->ip, clientSocket->port, buf);
//            log_debug("[%d] recv buf: {%s}\n", id, buf);
//            Write(fd, buf, ret);

            //TODO:正常来讲一次读入的数据应该是 message 的大小之和，尤其是在这种低负载的场景，所以没有实现缓冲机制，如果在高负载场景使用应该使用缓冲区
            decode_message(fd, ret);
        } else if (ret < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
            //读数据结束
//            log.print("[client " + clientSocket.ip + ":" + to_string(clientSocket.port) + "]\n{read end}");
//            printf("worker %d [%s : %d] sent end\n", clientSocket->idx, clientSocket->ip, clientSocket->port);
            break;
        } else {
            //连接断开
            epoll_del_fd2read(fd);
//            log.print("[client disconnect] " + clientSocket.ip + ":" + to_string(clientSocket.port));
//            clientAddr.erase(fd);
            struct Socket *sckt = NULL;
            for (int i = 0; i < terminalNum; ++i) {
                if (sockets[i].sockfd == fd) {
                    sckt = &sockets[i];
                    sckt->sockfd = 0;
                    break;
                }
            }
            log_trace("[%d] terminal {%s : %d} disconnect", id, sckt->ip, sckt->port);
            break;
        }
    }
}

/*
 * 主动连接成功
 * TODO: 在所有主动连接建立完成后通知初始控制终端
 */
void connect_success(int idx) {
    struct Socket *sckt = &sockets[idx];
    log_trace("[%d] connect terminal %d {%s : %d} successfully", id, idx, sckt->ip, sckt->port);
    encode_message(idx, 1, 0);
}

/*
 * 主动连接终端idx
 */
void try_connect(void* p) {
    int idx = (int)p;
    struct Socket *sckt = &sockets[idx];

    struct sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(sckt->port);
    int ret = inet_pton(AF_INET, sckt->ip, &addr.sin_addr.s_addr);
    if (ret == 0) {
        perror_exit("ip convert error");    //src不包含代表指定地址族中有效网络地址的字符串
    } else if (ret == -1){
        perror_exit("af to ip error");            //af不包含有效的地址族
    }

    ret = connect(sckt->sockfd, (struct sockaddr *)(&addr), sizeof(addr));
    if (ret == 0) {
        //连接成功
        connect_success(idx);
    } else if (errno != EINPROGRESS) {
        //连接建立失败，对端尚未开始监听，加入定时器事件
        log_trace("[%d] connect terminal %d {%s : %d} failure, retry after %d seconds", id, idx, sckt->ip, sckt->port, ALARM_TIME);
        alarmEvents[alarmEventNum].handler = try_connect;
        alarmEvents[alarmEventNum].p = p;
        ++alarmEventNum;
    } else {
        //连接正在建立
        epoll_add_fd2write(sckt->sockfd, true);
    }

}

/*
 * epoll_wait返回处理connect就绪事件
 */
void handle_connect(int sockfd) {
    //printf("handle_connect\n");
    epoll_del_fd2write(sockfd);

    int error = 0;
    socklen_t errorLen = sizeof(error);
    // 调用getsockopt来获取并清除sockfd上的错误
    if (getsockopt(sockfd, SOL_SOCKET, SO_ERROR, &error, &errorLen) < 0) {
        Close(sockfd);
        perror_exit("get socket option failed\n");
    }
    int idx = 0;
    for (int i = 0; i < active_connect_len; ++i) {
        if (sockets[active_connect[i]].sockfd == sockfd) {
            idx = active_connect[i];
            break;
        }
    }
    if (error != 0) {
        struct Socket *sckt = &sockets[idx];
        log_trace("[%d] connect terminal %d {%s : %d} failure, retry after %d seconds", id, idx, sckt->ip, sckt->port, ALARM_TIME);
        //连接失败，重新尝试连接
        alarmEvents[alarmEventNum].handler = try_connect;
        alarmEvents[alarmEventNum].p = (void *)idx;
        ++alarmEventNum;
        alarm(ALARM_TIME);
    } else {
        //连接成功
        connect_success(idx);
        epoll_add_fd2read(sockfd, true);
    }
}

/*
 * 处理epoll_wait返回事件
 * TODO: 处理被动建立连接一方关闭连接的错误
 */
void handle_events(int n) {
    for (int i = 0; i < n; ++i) {
        if (events[i].events & EPOLLIN) {
            int fd = events[i].data.fd;
            if (fd == sockets[id].sockfd) {
                accept_client(fd);
            } else if (fd == pipefd[0]) {
                handle_signal(fd);
            } else {
                read_buf(fd);
            }
        } else if (events[i].events & EPOLLOUT) {
            //连接事件
            int fd = events[i].data.fd;
            handle_connect(fd);
        }
    }
}

/*
 * 使用epoll监听连接socket
 */
void epoll_listen() {
    int lfd = Socket(AF_INET, SOCK_STREAM, 0);
    port_reuse(lfd);
//    set_nonblocking(lfd);
    easy_bind(lfd, sockets[id].port, sockets[id].ip);
    Listen(lfd, terminalNum);

    sockets[id].sockfd = lfd;

    epfd = Epoll_create(terminalNum + 5);
    epoll_add_fd2read(lfd, false);
}



/*
 * 使用epoll实现非阻塞connect
 */
void epoll_connect() {
    for (int i = 0; i < active_connect_len; ++i) {
        int sockfd = Socket(PF_INET, SOCK_STREAM, 0);
        int fdopt = set_nonblocking(sockfd);

        struct Socket *sckt = &sockets[active_connect[i]];
        sckt->sockfd = sockfd;

        try_connect((void *)active_connect[i]);

    }

    //如果加入定时器事件，开启定时后过一段时间重新连接
    if (alarmEventNum > 0) alarm(ALARM_TIME);
}

/*
 * 统一事件源，令所有的信号捕获函数都是向管道写入信息
 */
void sig_handler(int signo) {
    //保留原来的errno，保证函数的可重入性
    int old_errno = errno;
    Write(pipefd[1], &signo, 1);
    errno = old_errno;
}

/*
 * 使用epoll监听信号管道
 */
void epoll_signal() {
    //只能使用本地套接字
    check_error(socketpair(AF_UNIX, SOCK_STREAM, 0, pipefd), "socketpair error");
    set_nonblocking(pipefd[1]);
    epoll_add_fd2read(pipefd[0], false);
//    int sigs[5] = {SIGHUP, SIGCHLD, SIGTERM, SIGINT, SIGQUIT};
#define SIG_NUM 3
    int sigs[SIG_NUM] = {SIGHUP, SIGCHLD, SIGALRM};
    for (int i = 0; i < SIG_NUM; ++i) {
        int sig = sigs[i];
        struct sigaction act;
        memset(&act, 0, sizeof(act));
        act.sa_flags |= SA_RESTART;
        sigfillset(&act.sa_mask);
        act.sa_handler = sig_handler;
        check_error(sigaction(sig, &act, NULL), "sigaction error");
    }
}


void run() {
    if (confirm_cnt == active_connect_len) ++ready_cnt;

    epoll_listen();

    epoll_signal();

    epoll_connect();

    while (true) {
        int ret = Epoll_wait(epfd, events, MAX_EVENT_NUMBER, -1);
        handle_events(ret);
    }
}
